"""
Case Type   : 资源池化
Case Name   : 备机有业务场景下支持在线failover，极致RTO回放模式下，mv dss + kill dss
Create At   : 2025.04
Owner       : haomeng
Description :
    1、查看集群状态 cm_ctl query -Cv
    2、开启极致RTO参数
    3、主节点执行tpcc读写业务，备节点执行tpcc只读业务
    4、移除主节点dss路径,mv xxx/dss new_dir/，查看主节点dss进程并杀死进程，查看集群状态
       ps ux | grep 'dssserver' , cm_ctl query -Cv
    5、恢复原始dn路径,重启集群,恢复集群原始状态
        mv new_dir/dn xxx/,cm_ctl start
        cm_ctl switchover -a
Expect      :
    1、集群状态正常，数据库节点状态正常
    2、开启极致RTO成功
    3、业务执行成功
    4、移除主节点dss路径成功，集群状态异常,dss进程不存在,任意备节点升主，且升主节点未发生重启failover
    5、恢复原始dss路径成功,重启集群成功,恢复集群初始状态成功
History     :
"""

import os
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 == Primary_SH.get_node_num(), '单机环境不执行')
class SharedStorage(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.constant = Constant()
        self.com = Common()
        self.root_usernode = Node('Standby2Root')
        self.nodes = [Node(n) for n in ('PrimaryDbUser', 'Standby1DbUser', 'Standby2DbUser')]
        self.bak_path = os.path.join(os.path.dirname(macro.DSS_INSTANCE), 'dss_bak')

        self.param_list = ['recovery_parse_workers', 'recovery_redo_workers']
        self.value_list = ['4', '4']
        self.default_value = []

    def test_shared_storage(self):
        text = '-----step1:查看集群状态; expect:集群状态正常-----'
        self.log.info(text)

        status = Primary_SH.exec_cm_ctl(mode='query', param='-Cv')
        self.log.info(status)
        self.assertIn('cluster_state   : Normal', status, '执行失败' + text)

        text = '-----step2:开启极致RTO参数  expect:开启RTO参数成功-----'
        self.log.info(text)

        for param in self.param_list:
            value = Primary_SH.execut_db_sql(f'show {param}')
            self.default_value.append(value.splitlines()[-2].strip())
        self.log.info(self.default_value)

        for param in self.param_list:
            set_param = Primary_SH.execute_gsguc('set',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f'{param} = {self.value_list[self.param_list.index(param)]}')
            self.log.info(set_param)
            self.assertTrue(set_param, '参数设置失败')

        self.log.info('---重启集群使参数生效---')
        restart = Primary_SH.restart_db_cluster()
        self.log.info(restart)
        self.assertTrue(restart, '集群重启失败')

        self.log.info('---查看参数值修改是否成功---')
        for param in self.param_list:
            value = Primary_SH.execut_db_sql(f'show {param};')
            self.assertEqual(value.splitlines()[-2].strip(),
                             self.value_list[self.param_list.index(param)], '参数未修改成功')

        text = '-----step3:主节点执行读写业务，备节点执行只读业务  expect:启动业务成功-----'
        self.log.info(text)

        tpcc_cmd_p = f'sh runBenchmark.sh {macro.PRI_PROPS}'
        tpcc_thread_p = ComThread(self.com.startTPCC,
                                  args=(self.root_usernode,
                                        macro.TPCC_PATH,
                                        tpcc_cmd_p,))
        tpcc_thread_p.setDaemon(True)
        tpcc_thread_p.start()

        tpcc_cmd_s = f'sh runBenchmark.sh {macro.STA_PROPS}'
        tpcc_thread_s = ComThread(self.com.startTPCC,
                                  args=(self.root_usernode,
                                        macro.TPCC_PATH,
                                        tpcc_cmd_s,))
        tpcc_thread_s.setDaemon(True)
        tpcc_thread_s.start()

        text = '-----step4:移除主节点dss路径; expect:停止成功-----'
        self.log.info(text)

        mv_dss = f'''mv {macro.DSS_INSTANCE} {self.bak_path};
             ls {self.bak_path}'''
        dss_replace = self.nodes[0].sh(mv_dss).result()
        self.log.info(dss_replace)
        self.assertIn('cfg', dss_replace, '执行失败' + text)

        self.log.info('-----查看主节点dss进程并杀死进程,查看集群状态-----')
        dss_pid1 = []
        for n in range(3):
            pid = self.com.get_pid(self.nodes[n], 'dssserver')
            dss_pid1.append(pid)
        self.log.info(dss_pid1)

        self.nodes[0].sh(f'''kill -9 {dss_pid1[0]}''').result()

        time.sleep(120)

        sta1 = Primary_SH.exec_cm_ctl('query', '-Cv')
        self.log.info(sta1)
        self.assertIn('cluster_state   : Degraded', sta1, '执行失败' + text)
        self.assertIn('P Down    Manually stopped', sta1, '主节点正常')
        self.assertIn('S Primary Normal', sta1, '未有备节点升为主节点')

        dss_pid2 = []
        for n in range(1, 3):
            pid = self.com.get_pid(self.nodes[n], 'dssserver')
            dss_pid2.append(pid)
        self.assertEqual(dss_pid1[1], dss_pid2[0], 'dss进程发生改变')
        self.assertEqual(dss_pid1[2], dss_pid2[1], 'dss进程发生改变')

        tpcc_thread_p.join(70)
        p_res = tpcc_thread_p.get_result()
        self.log.info(p_res)

        tpcc_thread_s.join(70)
        s_res = tpcc_thread_s.get_result()
        self.log.info(s_res)
        self.assertIn(self.constant.TPCC_SUCCESS_MSG, p_res, '执行失败' + text)
        self.assertIn(self.constant.TPCC_SUCCESS_MSG, s_res, '执行失败' + text)

    def tearDown(self):
        text = '-----step5:恢复原始dss路径,重启集群,恢复集群初始状态; ' \
               'expect:启动成功,集群状态正常-----'
        self.log.info(text)

        remv_dss = f'''rm -rf {self.bak_path};
                       mv {self.bak_path} {macro.DSS_INSTANCE};
                       ls {macro.DSS_INSTANCE}'''
        dss_replace = self.nodes[0].sh(remv_dss).result()
        self.log.info(dss_replace)

        startdn = Primary_SH.exec_cm_ctl(mode='start')
        self.log.info(startdn)

        for param in self.param_list:
            reset_param = Primary_SH.execute_gsguc('set',
                                                   self.constant.GSGUC_SUCCESS_MSG,
                                                   f'{param} = {self.default_value[self.param_list.index(param)]}')
            self.log.info(reset_param)

        restart = Primary_SH.restart_db_cluster(False)
        self.log.info(restart)

        status = Primary_SH.exec_cm_ctl('query', '-Cv')
        self.log.info(status)

        time.sleep(60)
        recovery = Primary_SH.exec_cm_ctl(mode='switchover', param='-a')
        self.log.info(recovery)

        status = Primary_SH.exec_cm_ctl(mode='query', param='-Cv')
        self.log.info(status)

        for param in self.param_list:
            reset_value = Primary_SH.execut_db_sql(f'show {param};')
            self.log.info(reset_value)
            self.assertEqual(reset_value.splitlines()[-2].strip(),
                             self.default_value[self.param_list.index(param)], '参数恢复失败')

        self.assertIn('cfg', dss_replace, '恢复dss失败')
        self.assertIn(self.constant.cm_start_success_msg, startdn, '重启集群成功')
        self.assertIn(self.constant.cm_switchover_success_msg, recovery,
                      '执行失败' + text)
        self.assertIn('cluster_state   : Normal', status, '执行失败' + text)
        self.assertIn('P Primary Normal',
                      status.splitlines()[-1].split('|')[0].split('6001')[-1],
                      '执行失败' + text)
        self.assertIn('S Standby Normal',
                      status.splitlines()[-1].split('|')[1].split('6002')[-1],
                      '执行失败' + text)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
